// Copyright 2018 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// ------------------------------------------------------------------

#include <transport/ReceiverResource.h>
#include <transport/MessageReceiverInterface.h>
#include <cassert>
#include <elog/Log.hpp>

#define IDSTRING "(ID:" << std::this_thread::get_id() << ") " <<

using namespace std;

namespace vbs {
namespace transport {

ReceiverResource::ReceiverResource(TransportInterface& transport, const Locator_t& locator,
                                   uint32_t max_recv_buffer_size)
    : Cleanup(nullptr),
      LocatorMapsToManagedChannel(nullptr),
      mValid(false),
      mtx_(),
      cv_(),
      receiver(nullptr),
      max_message_size_(max_recv_buffer_size),
      active_callbacks_(0) {

    // Implementation functions are bound to the right transport parameters
    Cleanup = [&transport, locator]() {
        (void)transport.CloseInputChannel(locator);
    };
    LocatorMapsToManagedChannel = [&transport, locator](const Locator_t& locatorToCheck) -> bool {
        return (locator.kind == locatorToCheck.kind) && transport.DoInputLocatorsMatch(locator, locatorToCheck);
    };
}

ReceiverResource::ReceiverResource(ReceiverResource&& rValueResource) {
    std::lock_guard<std::mutex> guard(mtx_);

    Cleanup.swap(rValueResource.Cleanup);
    LocatorMapsToManagedChannel.swap(rValueResource.LocatorMapsToManagedChannel);
    receiver = rValueResource.receiver;
    rValueResource.receiver = nullptr;
    mValid = rValueResource.mValid;
    rValueResource.mValid = false;
    max_message_size_ = rValueResource.max_message_size_;
    active_callbacks_ = rValueResource.active_callbacks_.load();
    rValueResource.active_callbacks_ = 0;
}

bool ReceiverResource::SupportsLocator(const Locator_t& localLocator) {
    if (LocatorMapsToManagedChannel) {
        return LocatorMapsToManagedChannel(localLocator);
    }
    return false;
}

void ReceiverResource::RegisterReceiver(MessageReceiverInterface* rcv) {
    std::lock_guard<std::mutex> guard(mtx_);

    if (receiver == nullptr) {
        receiver = rcv;
    }
}

void ReceiverResource::UnregisterReceiver(MessageReceiverInterface* rcv) {
    std::lock_guard<std::mutex> guard(mtx_);

    if (receiver == rcv) {
        receiver = nullptr;
    }
}

void ReceiverResource::OnDataReceived(const unsigned char* data, const uint32_t len, const Locator_t& localLocator,
                                      const Locator_t& remoteLocator) {
    (void)localLocator;

    std::lock_guard<std::mutex> guard(mtx_);

    MessageReceiverInterface* rcv = receiver;

    if ((rcv != nullptr) && (active_callbacks_ >= 0)) {
        ++active_callbacks_;
        // TODO: Should we unlock in case UnregisterReceiver is called from callback ?
        rcv->processMsg(data, len, localLocator, remoteLocator);

        // allow disabling
        --active_callbacks_;
    }
}

void ReceiverResource::OnNewAccepted(const Locator_t& localLocator, const Locator_t& remoteLocator) {
    (void)localLocator;

    std::lock_guard<std::mutex> guard(mtx_);

    MessageReceiverInterface* rcv = receiver;

    if ((rcv != nullptr) && (active_callbacks_ >= 0)) {
        ++active_callbacks_;
        // TODO: Should we unlock in case UnregisterReceiver is called from callback ?
        rcv->processAccepted(localLocator, remoteLocator);

        // allow disabling
        --active_callbacks_;
    }
}

void ReceiverResource::OnDisconnected(const Locator_t& localLocator, const Locator_t& remoteLocator) {
    (void)localLocator;

    std::lock_guard<std::mutex> guard(mtx_);

    MessageReceiverInterface* rcv = receiver;

    if ((rcv != nullptr) && (active_callbacks_ >= 0)) {
        ++active_callbacks_;
        // TODO: Should we unlock in case UnregisterReceiver is called from callback ?
        rcv->processDisconnected(localLocator, remoteLocator);

        // allow disabling
        --active_callbacks_;
    }
}

void ReceiverResource::disable() {
    const std::chrono::milliseconds sleepInterval(10);

    if (Cleanup) {
        Cleanup();
    }

    while (active_callbacks_ > 0) {
        std::this_thread::sleep_for(sleepInterval);
    }
    // wait until all callbacks are finished
    std::lock_guard<std::mutex> guard(mtx_);
    // no more callbacks
    active_callbacks_ = -1;
}

ReceiverResource::~ReceiverResource() {}

}  // namespace transport
}  // namespace vbs
